package com.song.elasticsearch.config.kafka;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ResourceUtils;
import sun.security.provider.ConfigFile;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.URLDecoder;
import java.util.Properties;

@Data
@Configuration
@ConfigurationProperties(prefix = "kafka.server")
public class KafkaSecurity {
    private String kafkaSecurityModeFile;
    private String userKeyTableFile;
    private String krbConfFile;
    private String zookeeperServerPrincipal;
    private String userPrincipal;

    private String serverConfigPath;

    //安全认证
    public boolean securityValidate() {
        if (isSecurityModel()) {
            try {
                //!!注意，安全认证时，需要用户手动修改为自己申请的机机账号
                securityPrepare();
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }
        return false;
    }

    private Boolean isSecurityModel() {
        Boolean isSecurity = false;
        Properties securityProps = new Properties();
        kafkaSecurityModeFile = configPath() + kafkaSecurityModeFile;
        // file does not exist.
        if (!isExists(kafkaSecurityModeFile)) {
            return isSecurity;
        }
        try (FileInputStream fis = new FileInputStream(kafkaSecurityModeFile)) {
            securityProps.load(fis);
            if ("yes".equalsIgnoreCase(securityProps.getProperty("kafka.client.security.mode"))) {
                isSecurity = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isSecurity;
    }

    private void securityPrepare() throws Exception {
        userKeyTableFile = configPath() + userKeyTableFile;
        krbConfFile = configPath() + krbConfFile;
        KafkaLogin.setKrb5Config(krbConfFile);
        KafkaLogin.setZookeeperServerPrincipal(zookeeperServerPrincipal);
        KafkaLogin.setJaasFile(userPrincipal, userKeyTableFile);
        javax.security.auth.login.Configuration jaasConfig = javax.security.auth.login.Configuration.getConfiguration();
        for (Field field : jaasConfig.getClass().getDeclaredFields()) {
            if ("spi".equals(field.getName())) {
                field.setAccessible(true);
                Field modifiers = Field.class.getDeclaredField("modifiers");
                modifiers.setAccessible(true);
                field.setAccessible(true);
                Object spi = field.get(jaasConfig);
                if (spi instanceof ConfigFile.Spi) {
                    Method init = spi.getClass().getDeclaredMethod("init");
                    init.setAccessible(true);
                    init.invoke(spi);
                }
            }
        }
    }

    private String configPath() {
        String configPath = "";
        try {
            configPath = ResourceUtils.getURL("classpath:").getPath();
            configPath = URLDecoder.decode(configPath, "UTF-8");
            if (System.getProperty("os.name").toLowerCase().contains("windows")) {
                configPath = configPath.substring(1).replace("/", "\\");
            } else {
                configPath = serverConfigPath;
            }
        } catch (FileNotFoundException | UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return configPath;
    }

    public static Boolean isExists(String path){
        File file = new File(path);
        return file.exists();
    }
}
